/*
 * Copyright 2009-2010 Belmont Software Services
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.belmont.backup;

import java.util.*;

public class Producer implements IEventProducer, Runnable, IBackupConstants {
    static final int MAX_POOL_SIZE = 300;

    static Producer sharedInstance;
    Thread msgThread;
    Vector<IEventObserver>  observers = new Vector<IEventObserver>();
    Vector<ProducerMessage> messageQueue = new Vector<ProducerMessage>();
    Vector<ProducerMessage> messagePool = new Vector<ProducerMessage>();

    static class ProducerMessage {
	Object sender;
	int msg;
	Object arg;

	ProducerMessage(Object sender, int msg, Object arg) {
	    this.sender = sender;
	    this.msg = msg;
	    this.arg = arg;
	}
    }

    public static synchronized Producer getInstance() {
	if (sharedInstance == null) {
	    sharedInstance = new Producer();
	    sharedInstance.start();
	}

	return sharedInstance;
    }

    public synchronized void start() {
	if (msgThread == null) {
	    msgThread = new Thread(this);
	    msgThread.start();
	}
    }

    public synchronized void stop() {
	if (msgThread != null) {
	    msgThread.interrupt();
	    try {
		Thread.sleep(500);
	    } catch (InterruptedException ex) {
	    }
	    msgThread = null;
	}
    }

    public synchronized void removeObserver(IEventObserver client) {
	observers.removeElement(client);
    }

    public synchronized void addObserver(IEventObserver client, int start, int end) {
	removeObserver(client);
	observers.addElement(client);
    }

    public synchronized void sendNotify(Object sender, int msg, Object arg) {
	ProducerMessage m = null;
	int s = messagePool.size();
	/*
	while (s == MAX_POOL_SIZE) {
	    try {
		wait();
		System.out.println("WAIT FOR NOTIFY POOL");
	    } catch (InterruptedException ex) {
		return;
	    }
	}
	*/

	if (s > 0) {
	    m = messagePool.elementAt(0);
	    messagePool.removeElementAt(0);
	    m.sender = sender;
	    m.msg = msg;
	    m.arg = arg;
	} else {
	    m = new ProducerMessage(sender, msg, arg);
	}

	messageQueue.addElement(m);
	notifyAll();
    }

    void messageLoop() throws InterruptedException {
	while (! Thread.currentThread().isInterrupted()) {
	    try {
		Iterator<ProducerMessage> msgs = null;
		Vector<ProducerMessage> localq = null;
		
		synchronized (this) {
		    while (messageQueue.size() == 0) {
			wait();
		    }
		    localq = messageQueue;
		    messageQueue = new Vector<ProducerMessage>();
		}

		msgs = localq.iterator();

		while (msgs.hasNext()) {
		    ProducerMessage msg = msgs.next();
		    Iterator<IEventObserver> clients = observers.iterator();

		    while (clients.hasNext()) {
			IEventObserver observer = clients.next();
			try {
			    observer.notify(msg.sender, msg.msg, msg.arg);
			} catch (Throwable ex) {
			    // ignore exceptions from observer
			    Utils.log(LOG_ERROR, "Exception in observer notify ", ex);
			}
		    }
		}

		//
		// remove all the messages we sent from the queue
		//
		synchronized (this) {
		    int s = messagePool.size();
		    if (s < MAX_POOL_SIZE) {
			s = MAX_POOL_SIZE - s;
			int l = localq.size();
			int i;
			for (i=0; i < s && i < l; i++) {
			    messagePool.addElement(localq.elementAt(i));
			}
		    }
		}
	    } catch (Exception ex) {
		Utils.log(LOG_ERROR, "Message loop error ", ex);
	    }
	}
    }

    public void run() {
	try {
	    messageLoop();
	} catch (InterruptedException ex) {
	}
    }
}
